package com.atguigu.util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.util.Properties;

/**
 * @author hjy
 * @create 2023/3/24 10:05
 */
public class MyKafkaUtil_hjy {
    private static final String BOOTSTRAP_SERVER="47.102.112.46:9092";
    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupID){

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVER);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupID);
        return new FlinkKafkaConsumer<String>(topic, new DeserializationSchema<String>() {
            @Override
            public String deserialize(byte[] message) throws IOException {
                if (message==null){
                    return null;
                }
                return new String(message);
            }

            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return BasicTypeInfo.STRING_TYPE_INFO;
            }
        },props);
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic){

        return new FlinkKafkaProducer<String>(BOOTSTRAP_SERVER,
                topic,
                new SimpleStringSchema());
    }

    public static String getKafkaDDL(String topic,String groupId) {
        return "with(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = '"+topic+"',\n" +
                "    'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "',\n" +
                "    'properties.group.id' = '" + groupId + "',\n" +
                "    'scan.startup.mode' = 'group-offsets',\n" +
                "    'format' = 'json'\n" +
                ")";
    }

    public static String getTopicDBDDL(String groupId) {
        return "create table topic_db(\n" +
                "     `database` STRING,\n" +
                "     `table` STRING,\n" +
                "     `type` STRING,\n" +
                "     `data` MAP<STRING,STRING>,\n" +
                "     `old` MAP<STRING,STRING>,\n" +
                "     `pt` AS PROCTIME()\n" +
                ") " + getKafkaDDL(groupId);
    }
    public static String getKafkaSinkDDL(String topic) {
        return "with(\n" +
                "  'connector'='kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "',\n" +
                "  'format'='json'\n" +
                ")";
    }
    public static String getKafkaDDL(String groupId) {
        return "with(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'topic_db',\n" +
                "    'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "',\n" +
                "    'properties.group.id' = '" + groupId + "',\n" +
                "    'scan.startup.mode' = 'group-offsets',\n" +
                "    'format' = 'json'\n" +
                ")";
    }
}
